/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog.previous;

import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.aop.ProxyFactory;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.ipc.NRPC;
import com.nhncorp.neptune.common.ipc.NRPC.Server;
import com.nhncorp.neptune.common.util.NetworkUtil;
import com.nhncorp.neptune.tabletserver.ChangeLog;
import com.nhncorp.pleiades.client.LockEventData;
import com.nhncorp.pleiades.client.LockEventFilter;
import com.nhncorp.pleiades.client.LockEventMonitor;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.client.EventService.Events;
import com.nhncorp.pleiades.protocol.LockMetaData;

public class FileChangeLogServer implements FileChangeLogServerIF {
  public static final Log LOG = LogFactory.getLog(FileChangeLogServer.class.getName());

  //TabletName -> FileChangeLogManager
  private Map<String, FileChangeLogManager> logManagers = new HashMap<String, FileChangeLogManager>(100); 
  
  private NConfiguration conf;

  private String hostName;

  private Server server;
  
  private String lockPath;
  
  private FileChangeLogServerIF proxyObject;
  
  private boolean shutdowned = false;
  
  public static void main(String args[]) throws IOException {
    if(args.length > 0 && "-format".equals(args[0])) {
      formatAll();
      System.exit(0);
    }
    if(args.length < 1) {
      NConfiguration conf = new NConfiguration();
      String dataFilePath = conf.get("changelog.data.dir");
      if(dataFilePath == null || dataFilePath.length() == 0) {
        throw new IOException("no fs.append.dir value. check hadoop conf");
      }
      FileChangeLogServerIF changeLogServer = (FileChangeLogServerIF) ProxyFactory
          .createProxyClass("changeLogServer");
      changeLogServer.init(changeLogServer, conf);
      (new Thread(changeLogServer)).start();
    } else {
      int serverCount = Integer.parseInt(args[0]);
      for(int i = 0; i < serverCount; i++) {
        NConfiguration conf = new NConfiguration();
        String dataFilePath = conf.get("changelog.data.dir");
        if(dataFilePath == null || dataFilePath.length() == 0) {
          throw new IOException("no fs.append.dir value. check hadoop conf");
        }        
        conf.set("changelog.data.dir", dataFilePath + "/" + (i+1));
        conf.setInt("neptune.changelog.server.port", 57001 + i);
        
        FileChangeLogServer changeLogServer = (FileChangeLogServer) ProxyFactory
            .createProxyClass("changeLogServer");
        changeLogServer.init(changeLogServer, conf);
        (new Thread(changeLogServer)).start();
      }
    }
  }

  private static void formatAll() throws IOException {
    long startTime = System.currentTimeMillis();
    
    NConfiguration conf = new NConfiguration();
    //이전의 모든 lock 정보를 삭제할때 까지 반복 수행
    while(true) {
      try {
        Thread.sleep(5 * 1000);
      } catch (InterruptedException e1) {
        return;
      }
      
      LockService lockService = null;
      try {
        lockService = LockServiceFactory.getLockService(conf, "NeptuneMaster", true);
       
        if(lockService != null) {
          if(!lockService.exist(Constants.CHANGELOG_SERVER)) {
            //아무것도 없는 경우
            LOG.debug("no lock node");
            return;
          } else {
            boolean lockResult = lockService.deleteNode(Constants.CHANGELOG_SERVER, true);
            if(lockResult) {
              LOG.debug("deleted CHANGELOG_SERVER files and lock");
              lockService.releaseAll();
              return;
            } 
          }
        } else {
          LockServiceFactory.clear("NeptuneMaster");        
        }
        
        if(System.currentTimeMillis() - startTime > 30 * 1000) {
          LOG.error("Can't format. timeout");
          return;
        }
      } catch (Exception e) {
        LOG.error(e.getMessage(), e);
      } finally {
        if(lockService != null) {
          LockServiceFactory.close("NeptuneMaster");
        } else {
          LOG.fatal("Can't get LockServer: lock opwner=NeptuneMaster");
        }
      }
    }    
  }
  public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
    if (protocol.equals(FileChangeLogServerIF.class.getName())) {
      return FileChangeLogServerIF.versionID;
    } else {
      throw new IOException("Unknown protocol to change log server: " + protocol);
    }    
  }

  public void init(FileChangeLogServerIF changeLogServer,
    NConfiguration configuration) throws IOException {
    this.conf = configuration;
    this.proxyObject = changeLogServer;
    
    try {
      String dataFilePath = conf.get("changelog.data.dir");
      if(dataFilePath == null || dataFilePath.length() == 0) {
        throw new IOException("no fs.append.dir value. check hadoop conf");
      }
      //Class.forName("org.hsqldb.jdbcDriver");
    } catch (Exception e) {
      e.printStackTrace();
      throw new IOException(e.getMessage());
    }
    
    try {
      InetSocketAddress serverAddress = NetworkUtil.getAddress(InetAddress
          .getLocalHost().getHostName() + ":" + conf.getInt("neptune.changelog.server.port", 57001));
  
      this.hostName = serverAddress.getHostName() + ":" + serverAddress.getPort();
  
      lockPath = Constants.CHANGELOG_SERVER + "/" + 
                NetworkUtil.hostNameToFileName(hostName);
      
      LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
      LockMetaData lockMetaData = new LockMetaData(lockPath);
      lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
      
      int retry = 0;
      while(true) {
        boolean success = lockService.createNode(lockMetaData, true);
        
        if(success) {
          LOG.info("succeded lock:" + lockPath);
          break;
        } else {
          if(retry > 5) {
            LOG.fatal("Can't get server lock:" + lockPath + ". ChangeLogServer killed");
            System.exit(0);
          }
          LOG.warn("Can't get server lock:" + lockPath + ". but retry");
          Thread.sleep(1000);
          retry++;
        }
      }
      this.server = NRPC.getServer(lockService, proxyObject, serverAddress.getHostName(),
          serverAddress.getPort(), conf.getInt("changeLogServer.handler.count",
              10), false, conf);
    } catch (Exception e) {
      e.printStackTrace();
      throw new IOException(e.getMessage());
    }
    
    LOG.info("Change log server started:" + hostName);
    try {
      addLockEventHandler();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  
  public void run() {
    try {
      this.server.start();
      this.server.join();
    } catch(Exception e) {
      e.printStackTrace(System.out);
      return;
    }
  }
  
  private void addLockEventHandler() throws IOException {
    LockEventMonitor showdownMonitor = new LockEventMonitor() {
      public LockEventFilter getFilter() {
        LockEventFilter filter = new LockEventFilter();
        filter.addConstraint(LockEventFilter.Operator.PF, lockPath);
        
        List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
        lockEvents.add(LockService.Events.SESSION_EXPIRED);
        lockEvents.add(LockService.Events.NODE_REMOVED);
        filter.setLockEventList(lockEvents);
        
        return filter;
      }
  
      public void notifyEvent(LockService.Events eventType, LockEventData eventData) {  
        try {
          LOG.warn("ChangeLogServer lock expired:" + eventData.getLockId());
          checkForShutdown();
        } catch (Exception e) {
          LOG.error(e);
        }
      }
    };
    
    LOG.info("Added LockEventHandler:" + lockPath);
    if(LockServiceFactory.getLockService(conf, hostName, true).getEventHandler() != null) {
      LockServiceFactory.getLockService(conf, hostName, true).getEventHandler().addLockEventMonitor(showdownMonitor);
    }
  }
  
  private void checkForShutdown() {
    //Session이 expire되는 경우 네트워크 문제 등으로 인해 Pleiades에 lock을 유지 못하는 경우이다.
    //lock 파일이 존재하지 않으면 종료
    try {
      LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
      if(lockService.getMetaData(lockPath) == null) {
        LOG.fatal("ChangeLogServer stopped. session expired:" + lockPath);
        shutdown();
        return;
      }
  
      int retry = 0;
      
      while(retry < 5) {
        //lock을 다시 획득할 수 있으면 정상적으로 계속 수행
        if(lockService.getLock(lockPath, LockService.LockMode.LOCK_EXCLUSIVE)) {
          return;
        }
        try {
          Thread.sleep(1*1000);
        } catch (InterruptedException e) {
          return;
        }
        retry++;
      }
      LOG.fatal("ChangeLogServer stopped. session expired:" + lockPath);
      shutdown();
    } catch (IOException e) {
      LOG.fatal("ChangeLogServer stopped. session expired:" + e.getMessage(), e);
      shutdown();
    }
  }
  
  public String addChangeLog(String tabletName, String txId, int seq, ChangeLog changeLog) throws IOException {
    //LOG.debug("addChangeLog:" + tabletName);
    FileChangeLogManager logManager = null;
    synchronized(logManagers) {
      logManager = logManagers.get(tabletName);
      if(logManager == null) {
        logManager = new FileChangeLogManager(conf, tabletName, hostName);
        logManagers.put(tabletName, logManager);
      }
    }
    
    return logManager.addChangeLog(txId, seq, changeLog);
  }
  
  public void close(String tabletName, boolean writable) throws IOException {
    FileChangeLogManager logManager = null;
    synchronized(logManagers) {
      logManager = logManagers.get(tabletName);
      if(logManager == null) {
        logManager = new FileChangeLogManager(conf, tabletName, hostName);
        logManagers.put(tabletName, logManager);
      }
    }
    logManager.close(writable);
  }
  
  public boolean startMinorCompaction(String tabletName) throws IOException {
    //LOG.debug("Begin startMinorCompaction:" + hostName + "," + tabletName);
    FileChangeLogManager logManager = null;
    synchronized(logManagers) {
      logManager = logManagers.get(tabletName);
      if(logManager == null) {
        logManager = new FileChangeLogManager(conf, tabletName, hostName);
        logManagers.put(tabletName, logManager);
      }
    }
    boolean result = logManager.rollingLogFile(tabletName);
    if(!result) {
      LOG.error("Can't new chnage log file for minorCompaction:" + hostName + "," + tabletName);
    }
    //LOG.debug("End startMinorCompaction:" + hostName + "," + tabletName);
    return result;
  }

  public boolean endMinorCompaction(String tabletName) throws IOException {
    FileChangeLogManager logManager = null;
    synchronized(logManagers) {
      logManager = logManagers.get(tabletName);
      if(logManager == null) {
        logManager = new FileChangeLogManager(conf, tabletName, hostName);
        logManagers.put(tabletName, logManager);
      }
    }
    boolean result = logManager.deleteOldFile();
    if(!result) {
      LOG.error("Can't complete minorCompaction:" + hostName + "," + tabletName);
    }
    //LOG.debug("End startMinorCompaction:" + hostName + "," + tabletName);
    return result;
  }

  public void format() throws IOException {
    synchronized(logManagers) {
      logManagers.clear();
    }
    String path = conf.get("changelog.data.dir");
    File file = new File(path);
    File[] files = file.listFiles();
    if(files == null || files.length == 0) {
      return;
    }
    
    for(File eachFile: files) {
      String tabletName = eachFile.getName();
      FileChangeLogManager logManager = new FileChangeLogManager(conf, tabletName, hostName);
      logManager.delete();
    }
  }

  public String getHostName() {
    return hostName;
  }

  public ChangeLogStatus getChangeLogStatus(String tabletName) throws IOException {
    FileChangeLogManager logManager = null;
    synchronized(logManagers) {
      logManager = logManagers.get(tabletName);
      if(logManager == null) {
        logManager = new FileChangeLogManager(conf, tabletName, hostName);
        logManagers.put(tabletName, logManager);
      }
    }    
    return logManager.getChangeLogStatus();
  }

  public boolean isShutdowned() {
    return shutdowned;
  }

  /**
   * ChangeLog 파일로 부터 ChangeLog 레코드를 읽는다.
   * MinorCompaction 처리 중에 kill 된 경우에는 파일이 2개이기 때문에 2개 모두 처리한다.
   * 순서는 OLD -> NEW
   * @param tabletName
   * @return
   * @throws IOException
   */
  public ChangeLog nextChangeLog(String tabletName) throws IOException {
    FileChangeLogManager logManager = null;
    synchronized(logManagers) {
      logManager = logManagers.get(tabletName);
      if(logManager == null) {
        logManager = new FileChangeLogManager(conf, tabletName, hostName);
        logManagers.put(tabletName, logManager);
      }
    } 
    return logManager.nextChangeLog();
  }

  public void delete(String tabletName) throws IOException {
    FileChangeLogManager logManager = null;
    synchronized(logManagers) {
      logManager = logManagers.get(tabletName);
      if(logManager == null) {
//        logManager = new FileChangeLogManager(conf, tabletName, hostName);
//        logManagers.put(tabletName, logManager);
        return;
      } 
    } 
    logManager.delete();
    synchronized(logManagers) {
      logManagers.remove(tabletName);
    }
  }
  
  public void shutdown() {
    LOG.fatal("shutdown changelogserver:" + hostName);
    if(shutdowned) {
      return;
    }

    if(!conf.getBoolean("testmode", false)) {
      LOG.fatal("shutdown changelogserver:" + hostName);
      Thread.dumpStack();
      System.exit(0);
    } 
    
    try {
      LockServiceFactory.close(hostName);
    } catch (Exception e) {
    }
    
    LOG.fatal("server.stop:" + hostName);
    server.stop();
    synchronized(logManagers) {
      for(FileChangeLogManager eachLogManager: logManagers.values()) {
        eachLogManager.close();
      }
    }

    LockServiceFactory.close(hostName);
    
    LOG.fatal("shutdown changelogserver:" + hostName);
    
    shutdowned = true;
  }

  public String test() {
    return "Y";
  }

  public String getTestHandlerKey() {
    return hostName;
  }
  

  /**
   * only for test case
   * @throws IOException
   */
  public static void doFormat() throws IOException {
    LockService lockService = null;

    boolean lockResult = true;
    long startTime = System.currentTimeMillis();
    while(true) {
      try {
        Thread.sleep(2 * 1000);
        lockService = LockServiceFactory.getLockService(new NConfiguration(), "ChangeLogServer", false);
        if(lockService !=  null) {
          if(lockService.exist(Constants.CHANGELOG_SERVER)) {
            lockResult = lockService.deleteNode(Constants.CHANGELOG_SERVER, true);
          } 
          if(lockResult) {
            return;
          }
          LockServiceFactory.clear("ChangeLogServer");
        } else {
          LockServiceFactory.clear("ChangeLogServer");
        }
        
        if(System.currentTimeMillis() - startTime > 30 * 1000) {
          throw new IOException("Can't delete all locks. timeout");
        }
      } catch (Exception e) {
        LOG.error(e);
        //e.printStackTrace(System.out);
      } finally {
        if(lockService != null) {
          lockService.releaseAll();
          LockServiceFactory.close("ChangeLogServer");
        }
        LOG.debug("delete all changelogs");
      }
    }
  }
}
